
package net.lab0.fractal.concurrent;


import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import net.lab0.fractal.exception.IllegalOperationException;
import net.lab0.tools.Pair;


/**
 * @author  116
 */
public class QueueableRunnableGroup<In, Out>
{
    /**
	 * @uml.property  name="newQueueSize"
	 */
    private static int                                               newQueueSize = 100;
    
    /**
	 * @uml.property  name="in"
	 */
    private BlockingQueue<In>                                        in;
    /**
	 * @uml.property  name="out"
	 */
    private BlockingQueue<Out>                                       out;
    
    /**
	 * @uml.property  name="previousGroup"
	 * @uml.associationEnd  
	 */
    private QueueableRunnableGroup<?, In>                            previousGroup;
    /**
	 * @uml.property  name="nextGroup"
	 * @uml.associationEnd  
	 */
    private QueueableRunnableGroup<Out, ?>                           nextGroup;
    
    private int                                                      running      = 0;
    
    protected Set<Pair<QueueableRunnableInterface<In, Out>, Thread>> queueables;
    
    public QueueableRunnableGroup()
    {
        queueables = new HashSet<Pair<QueueableRunnableInterface<In, Out>, Thread>>();
    }
    
    public QueueableRunnableGroup(QueueableRunnableGroup<?, In> previousGroup, QueueableRunnableGroup<Out, ?> nextGroup)
    throws IllegalOperationException
    {
        this();
        this.previousGroup = previousGroup;
        this.nextGroup = nextGroup;
        this.setInGroup(previousGroup);
        this.setOutGroup(nextGroup);
    }
    
    public void setOutGroup(QueueableRunnableGroup<Out, ?> queuedThreadGroup)
    throws IllegalOperationException
    {
        if (queuedThreadGroup.getIn() != null)
        {
            this.out = queuedThreadGroup.getIn();
        }
        else
        {
            LinkedBlockingQueue<Out> queue = new LinkedBlockingQueue<Out>(newQueueSize);
            this.setOut(queue);
            queuedThreadGroup.setInGroup(this);
        }
        nextGroup = queuedThreadGroup;
    }
    
    public void setInGroup(QueueableRunnableGroup<?, In> queuedThreadGroup)
    throws IllegalOperationException
    {
        if (queuedThreadGroup.getOut() != null)
        {
            this.in = queuedThreadGroup.getOut();
        }
        else
        {
            LinkedBlockingQueue<In> queue = new LinkedBlockingQueue<In>(newQueueSize);
            this.setIn(queue);
            queuedThreadGroup.setOutGroup(this);
        }
        previousGroup = queuedThreadGroup;
    }
    
    /**
	 * @return
	 * @uml.property  name="in"
	 */
    public BlockingQueue<In> getIn()
    {
        return in;
    }
    
    /**
	 * @param in
	 * @throws IllegalOperationException
	 * @uml.property  name="in"
	 */
    protected void setIn(BlockingQueue<In> in)
    throws IllegalOperationException
    {
        if (this.in != null)
        {
            throw new IllegalOperationException("Input is allready set");
        }
        if (in == null)
        {
            throw new IllegalOperationException("Input can't be null");
        }
        this.in = in;
    }
    
    /**
	 * @return
	 * @uml.property  name="out"
	 */
    public BlockingQueue<Out> getOut()
    {
        return out;
    }
    
    /**
	 * @param out
	 * @throws IllegalOperationException
	 * @uml.property  name="out"
	 */
    protected void setOut(BlockingQueue<Out> out)
    throws IllegalOperationException
    {
        if (this.out != null)
        {
            throw new IllegalOperationException("Output is allready set");
        }
        if (out == null)
        {
            throw new IllegalOperationException("Output can't be null");
        }
        this.out = out;
    }
    
    /**
	 * @return
	 * @uml.property  name="newQueueSize"
	 */
    public static int getNewQueueSize()
    {
        return newQueueSize;
    }
    
    /**
	 * @param newQueueSize
	 * @uml.property  name="newQueueSize"
	 */
    public static void setNewQueueSize(int newQueueSize)
    {
        QueueableRunnableGroup.newQueueSize = newQueueSize;
    }
    
    public void startWorkers()
    {
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            int threadNumber = 0;
            if (p.b == null)
            {
                Thread t = new Thread(p.a, "" + getClass() + " #" + threadNumber);
                threadNumber++;
                p.b = t;
                t.start();
                // System.out.println(getClass() + " : start " + p.a + "(thread " + p.b + ")");
            }
        }
    }
    
    public void stopWorkers()
    {
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            p.a.stopWork();
            // System.out.println("stop : " + p.a);
        }
    }
    
    public void joinAll()
    throws InterruptedException
    {
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            if (p.b != null)
            {
                p.b.join();
                
                // System.out.println("joined : " + p.b);
            }
        }
    }
    
    /**
     * 
     * @return <code>true</code> if all workers of this groups are stopped
     */
    public boolean isStopped()
    {
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            if (!p.a.isStopped())
            {
                return false;
            }
        }
        return true;
    }
    
    public synchronized void addRunnable(QueueableRunnableInterface<In, Out> q)
    {
        queueables.add(new Pair<QueueableRunnableInterface<In, Out>, Thread>(q, null));
        running++;
    }
    
    /**
     * Must be called when a worker finishes its <code>run()</code> method
     * 
     * @param q
     *            the {@link QueueableRunnableInterface} which finished its job
     */
    public synchronized void notifyRunnableFinished(QueueableRunnableInterface<In, Out> q)
    {
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            if (p.a.equals(q) && p.b != null)
            {
                p.a.stopWork();
                running--;
            }
        }
        
        if (running == 0 && nextGroup != null)
        {
            nextGroup.notifyPreviousGroupFinished();
        }
    }
    
    private void notifyPreviousGroupFinished()
    {
        // System.out.println(getClass() + " : finish notification");
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            p.a.finishWork();
        }
    }
    
    /**
	 * @return
	 * @uml.property  name="previousGroup"
	 */
    public QueueableRunnableGroup<?, In> getPreviousGroup()
    {
        return previousGroup;
    }
    
    /**
	 * @return
	 * @uml.property  name="nextGroup"
	 */
    public QueueableRunnableGroup<Out, ?> getNextGroup()
    {
        return nextGroup;
    }
    
    protected Set<QueueableRunnableInterface<In, Out>> getRunnables()
    {
        Set<QueueableRunnableInterface<In, Out>> ret = new HashSet<QueueableRunnableInterface<In, Out>>(queueables.size());
        for (Pair<QueueableRunnableInterface<In, Out>, Thread> p : queueables)
        {
            ret.add(p.a);
        }
        return ret;
    }
}
